package cn.doitedu.demo.functions;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

import java.text.ParseException;
import java.time.LocalDateTime;
import java.util.Date;

public class 自定义函数示例1 {

    public static void main(String[] args) {
        // 构造编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 注册自定义函数
        tenv.createTemporaryFunction("mask_name",MaskNameUDF.class);
        tenv.createTemporaryFunction("time_trunc",MyTimeTrunc.class);

        // 调用自定义函数
        //tenv.executeSql("select mask_name('欧阳锋'),mask_name('欧阳娜娜'),mask_name('古龙')").print();
        tenv.executeSql("select " +
                "time_trunc('2024-01-12 11:36:48',60)," +
                "time_trunc('2024-01-12 11:36:48',30)," +
                "time_trunc('2024-01-12 11:36:48',10)," +
                "time_trunc('2024-01-12 11:36:48',5)," +
                "time_trunc('2024-01-12 11:36:48',1)").print();




    }


    /**
     * 自定义函数的实现
     */
    public static class MaskNameUDF extends ScalarFunction {
        public String eval(String name){

            String first = name.substring(0, 1);
            String last = "*";
            if(name.length()>2){
                last = name.substring(name.length()-1,name.length());
                String mid="";
                for(int i = 0;i<name.length()-2;i++){
                    mid += "*";
                }
                last = mid+last;
            }

            return first+last;
        }

    }

    /**
     * 自定义函数的实现
     * 给定一个时间， 2024-01-12 10:26:30
     * 按照指定的粒度进行取整：
     * 比如，按 30分钟取整 => 2024-01-12 10:00:00
     * 比如，按 10分钟取整 => 2024-01-12 10:30:00
     * 比如，按 5分钟取整 => 2024-01-12 10:25:00
     * 比如，按 1分钟取整 => 2024-01-12 10:26:00
     *
     */
    public static class MyTimeTrunc extends ScalarFunction{

        public String eval(String originTime, int interval){
            String truncated = null;
            try {
                Date date = DateUtils.parseDate(originTime, "yyyy-MM-dd HH:mm:ss");
                long time = date.getTime();

                //time = time - time%(interval*60*1000L);
                time  = (time / (interval*60*1000L))*(interval*60*1000L);


                truncated = DateFormatUtils.format(time, "yyyy-MM-dd HH:mm:ss");
            } catch (ParseException e) {
                throw new RuntimeException(e);
            }


            return truncated;
        }

    }



}
